package com.diao.flink.streamapi.source

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

import org.apache.flink.streaming.api.scala._

import scala.util.Random
/**
 * @author: Chenzhidiao
 * @date: 2020/3/26 14:46
 * @description:
 * @version: 1.0
 */
object MySource {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val stream4: DataStream[SensorReading] = env.addSource(new SensorSource)

    stream4.print()
    env.execute()
  }
}
class SensorSource extends SourceFunction[SensorReading] {
  //定义一个标记，决定数据源是否继续
  var flag: Boolean = true

  //继续生成数据
  override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = {
    val rand = new Random()
    //初始化定义一组传感器数据
    var curTemp = 1.to(10).map(
      i => ("sensor_" + i, 60 + rand.nextGaussian() * 20)
    )

    //产生数据流
    while (flag) {
      //在前一次的基础上更新温度值
      curTemp.map(t => (t._1, t._2 + rand.nextGaussian()))
      //获取当前时间戳
      val timestamp = System.currentTimeMillis()
      curTemp.foreach(
        t => sourceContext.collect(SensorReading(t._1, timestamp, t._2))
      )
      //设置时间间隔
      Thread.sleep(500)

    }
  }

  //取消数据生成
  override def cancel(): Unit = {
    flag = false
  }
}